package com.kuang.utils;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * MQTT工具类操作
 *
 * @author zoe_zls
 * @since v1.1.0 2024-03-4
 */
@Slf4j
@Service
public class MQTTConnect {

    //private String HOST = "tcp://www.thxxsd.top:1883"; //mqtt服务器的地址和端口号
    @Value("${mqtt.host}")
    private String HOST;
    @Value("${mqtt.clientid}")
    private String clientId;
    private MqttClient mqttClient;
 
    /**
     * 客户端connect连接mqtt服务器
     *
     * @param userName     用户名
     * @param passWord     密码
     * @param mqttCallback 回调函数
     **/
    public void setMqttClient(String userName, String passWord, MqttCallback mqttCallback) throws MqttException {
        MqttConnectOptions options = mqttConnectOptions(userName, passWord);
        if (mqttCallback == null) {
            mqttClient.setCallback(new Callback());
        } else {
            mqttClient.setCallback(mqttCallback);
        }
        mqttClient.connect(options);
    }
 
    /**
     * MQTT连接参数设置
     */
    private MqttConnectOptions mqttConnectOptions(String userName, String passWord) throws MqttException {
        mqttClient = new MqttClient(HOST, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(userName);
        options.setPassword(passWord.toCharArray());
        /*
        在MQTT（Message Queuing Telemetry Transport）中，设置连接超时可以确保客户端在特定的时间内等待连接到服务器的响应。
        如果在指定的时间内没有收到服务器的响应，客户端应该取消连接尝试并且可以选择重新连接。
        */
        options.setConnectionTimeout(10);///默认：30 设置超时时间 单位为秒
        options.setAutomaticReconnect(true);//默认：false --true 自动重连 false  不自动重连
        options.setCleanSession(false);//默认：true 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
        //options.setKeepAliveInterval(20);//默认：60  // 设置会话心跳时间单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        return options;
    }
 
    /**
     * 关闭MQTT连接
     */
    public void close() throws MqttException {
        mqttClient.disconnect();
        mqttClient.close();
    }
 
    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public boolean pub(String topic, String msg) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage();
        //mqttMessage.setQos(2);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
        boolean retained = mqttMessage.isRetained();
        return retained;
    }
 
    /**
     * 向某个主题发布消息
     *
     * @param topic: 发布的主题
     * @param msg:   发布的消息
     * @param qos:   消息质量    Qos：0、1、2
     */
    public void pub(String topic, String msg, int qos) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
    }
 
    /**
     * 订阅某一个主题 ，此方法默认的的Qos等级为：1
     *
     * @param topic 主题
     */
    public void sub(String topic) throws MqttException {
        mqttClient.subscribe(topic);
    }
 
    /**
     * 订阅某一个主题，可携带Qos
     *
     * @param topic 所要订阅的主题
     * @param qos   消息质量：0、1、2
     */
    public void sub(String topic, int qos) throws MqttException {
        mqttClient.subscribe(topic, qos);
    }

    /**
     * 订阅多个主题，可携带Qos
     *订阅某一个主题 ，此方法默认的的Qos等级为：1
     * @param topic 所要订阅的主题
     */
    public void subAll(String[] topic) throws MqttException {
        mqttClient.subscribe(topic);
    }

    /**
     * 订阅多个主题，可携带Qos
     * @param topic 所要订阅的主题
     */
    public void subAll(String[] topic,int[] qos) throws MqttException {
        mqttClient.subscribe(topic,qos);
    }
 
    /**
     * main函数自己测试用
     */
//    public static void main(String[] args) throws MqttException, JsonProcessingException {
//        MQTTConnect mqttConnect = new MQTTConnect();
//        mqttConnect.setMqttClient("zls", "123", new Callback());
//        //mqttConnect.sub("com/iot/init");//订阅主题
//        //mqttConnect.pub("com/iot/init", "Mr.Qu" + (int) (Math.random() * 100000000));//发布主题
//        mqttConnect.sub("hxl");//订阅主题
//        Map<String,Object> map = new LinkedHashMap();
//        map.put("status",1);
//        map.put("rCode","000000000");
//        map.put("aEle",0.0);
//        map.put("aWater",0.0);
//        map.put("dep",0.0);
//        map.put("pre",0.0);
//        map.put("mode",1);
//        map.put("price","单价");
//        map.put("runStatus",2);
//        map.put("lat","百度坐标");
//        map.put("lon","百度坐标");
//        map.put("cardType",3);
//        map.put("cardNo","");
//        ObjectMapper objectMapper = new ObjectMapper();
//        String s = objectMapper.writeValueAsString(map);
//        //System.out.println(s);
//        mqttConnect.pub("zls", s);//发布主题
//    }
}